package com.flink.job;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.FlatMapIterator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.Iterator;

/**
 * 借助netcat向本机指定端口发送文本信息，而后通过Flink实时接收，并输出个数在控制台
 */
public class NetCat {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置监控数据流时间间隔（官方叫状态与检查点）
        see.enableCheckpointing(5000);
        DataStreamSource<String> data = see.socketTextStream("localhost", 8888);
        data.flatMap(new FlatMapIterator<String, String>() {
            @Override
            public Iterator<String> flatMap(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return new Tuple2<>(s, 1);
            }
        }).keyBy(0).sum(1).print();
        see.execute();
    }
}

